ROS2 动作通信

动作概念

1. 什么是动作（Action）

在 ROS 2 中，**动作（Action）**是一种用于执行长时间运行任务的通信机制。它结合了话题和服务的特点，提供了一种更灵活的异步通信方式。

动作适合处理以下类型的任务：

机器人导航到目标位置（可能需要几秒到几分钟）
机械臂执行复杂的运动轨迹
执行需要持续反馈的长时间计算任务
需要能够被中途取消的任务

说明动作的核心特点
目标（Goal）：客户端发送的任务目标
反馈（Feedback）：服务端定期发送的任务进度信息
结果（Result）：任务完成后返回的最终结果

2. Action 与 Topic、Service 的区别

话题（Topic）

单向、异步通信
持续数据流
发布者-订阅者模式
无请求-响应机制
适合传感器数据流

服务（Service）

双向、同步通信
一次请求-一次响应
客户端-服务端模式
阻塞式等待结果
适合快速计算任务

动作（Action）

双向、异步通信
目标-反馈-结果
客户端-服务端模式
可取消、可监控进度
适合长时间任务

text

客户端发送目标 → 服务端执行任务 ↔ 持续发送反馈
返回最终结果

3. Action 的客户端与服务端

在 ROS 2 动作通信中：

Action Server（服务端）：负责接收目标请求，执行任务，定期发布反馈，返回最终结果
Action Client（客户端）：负责发送目标请求，接收反馈和结果，可以取消正在执行的任务

text

客户端（发送目标请求） → 服务端（接收并执行）
服务端（定期发布反馈） → 客户端（接收进度更新）
服务端（返回最终结果） → 客户端（处理结果）

4. .action 文件结构

动作接口通过 .action 文件定义，包含三个部分：

text

# 目标（Goal）部分 - 客户端发送给服务端的请求数据
int64 target_count
---
# 结果（Result）部分 - 任务完成后返回的数据
int64 final_count
bool success
string message
---
# 反馈（Feedback）部分 - 任务执行过程中的进度信息
int64 current_count
float64 progress

.action 文件使用 --- 分隔符来区分三个部分：

第一部分（Goal）：定义目标请求数据
第二部分（Result）：定义最终结果数据
第三部分（Feedback）：定义实时反馈数据

学习内容：动作实践

（一）创建 Action 接口包

首先创建一个用于存放自定义动作接口的功能包：

bash

cd ~/ros2_ws/src
ros2 pkg create --build-type ament_cmake action_interfaces

说明注意
自定义接口包通常使用 ament_cmake 构建类型，而不是 ament_python。

（二）创建 action 目录

在功能包根目录下创建 action 目录：

bash

cd action_interfaces
mkdir action

（三）编写 Countdown.action 自定义动作接口

在 action 目录下创建 Countdown.action 文件：

text

# action/Countdown.action
# Goal - 目标：设置倒计时的起始数字
int64 target_number
---
# Result - 结果：倒计时完成后的信息
bool success
string message
---
# Feedback - 反馈：当前倒计时的数字
int64 current_number

（四）修改 CMakeLists.txt 和 package.xml

1. 修改 CMakeLists.txt

在 CMakeLists.txt 中添加动作接口的编译配置：

cmake

cmake_minimum_required(VERSION 3.8)
project(action_interfaces)

if(CMAKE_COMPILER_IS_GNUCXX OR CMAKE_CXX_COMPILER_ID MATCHES "Clang")
  add_compile_options(-Wall -Wextra -Wpedantic)
endif()

find_package(ament_cmake REQUIRED)
find_package(rosidl_default_generators REQUIRED)

# 添加动作接口文件的声明
rosidl_generate_interfaces(${PROJECT_NAME}
  "action/Countdown.action"
)

# 添加依赖
ament_export_dependencies(rosidl_default_runtime)

ament_package()

2. 修改 package.xml

在 package.xml 中添加必要的依赖：

xml

<?xml version="1.0"?>
<package format="3">
  <name>action_interfaces</name>
  <version>0.0.0</version>
  <description>Custom action interfaces package</description>
  <maintainer email="user@todo.todo">user</maintainer>
  <license>TODO: License declaration</license>

  <buildtool_depend>ament_cmake</buildtool_depend>

  <build_depend>rosidl_default_generators</build_depend>
  <exec_depend>rosidl_default_runtime</exec_depend>
  <member_of_group>rosidl_interface_packages</member_of_group>

  <export>
    <build_type>ament_cmake</build_type>
  </export>
</package>

（五）创建使用动作接口的 Python 功能包

创建一个新的 Python 功能包来实现动作的客户端和服务端：

bash

cd ~/ros2_ws/src
ros2 pkg create --build-type ament_python learn_action --dependencies rclpy action_interfaces

（六）编写 Action Server 节点

在 learn_action 功能包下创建 action_server.py 文件：

python

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

import rclpy                                       # ROS2 Python接口库
from rclpy.node import Node                        # ROS2 节点类
from rclpy.action import ActionServer              # ROS2 动作服务端类
from rclpy.action.server import ServerGoalHandle   # 目标处理句柄
from rclpy.callback_groups import ReentrantCallbackGroup  # 回调组，允许多线程回调

# 导入自定义动作接口
from action_interfaces.action import Countdown

import time                                        # 时间模块，用于倒计时延迟

class CountdownActionServer(Node):
    """
    倒计时动作服务端节点
    接收目标数字，从该数字开始倒计时到0，期间发送反馈
    """

    def __init__(self, node_name='countdown_action_server'):
        super().__init__(node_name)

        # 创建回调组，允许多个回调同时执行（重要：防止阻塞）
        self._callback_group = ReentrantCallbackGroup()

        # 创建动作服务端
        # 参数：动作类型、动作名称、执行回调函数
        self._action_server = ActionServer(
            self,
            Countdown,                          # 动作接口类型
            'countdown',                        # 动作名称
            self.execute_callback,              # 执行回调函数
            callback_group=self._callback_group # 指定回调组
        )

        self.get_logger().info('倒计时动作服务端已启动！')

    def execute_callback(self, goal_handle: ServerGoalHandle):
        """
        动作执行回调函数
        当收到目标请求时被调用，执行倒计时逻辑

        参数:
            goal_handle: 目标句柄，包含请求数据和控制方法
        """
        self.get_logger().info('收到倒计时目标请求...')

        # 获取目标数据
        target_number = goal_handle.request.target_number

        # 验证目标数字是否有效
        if target_number <= 0:
            # 如果目标无效，中止任务
            goal_handle.abort()
            result = Countdown.Result()
            result.success = False
            result.message = f'目标数字无效: {target_number}，必须大于0'
            return result

        # 目标有效，开始执行
        self.get_logger().info(f'开始倒计时，从 {target_number} 到 0')

        # 创建反馈消息对象
        feedback_msg = Countdown.Feedback()

        # 创建结果消息对象
        result = Countdown.Result()

        # 执行倒计时循环
        current_number = target_number
        while current_number >= 0:
            # 检查是否被客户端取消
            if goal_handle.is_cancel_requested:
                goal_handle.canceled()
                self.get_logger().info('目标被取消')
                result.success = False
                result.message = '倒计时被取消'
                return result

            # 发送反馈
            feedback_msg.current_number = current_number
            goal_handle.publish_feedback(feedback_msg)
            self.get_logger().info(f'反馈: 当前数字 {current_number}')

            # 倒计时减一
            current_number -= 1

            # 延迟1秒，模拟任务执行
            time.sleep(1)

        # 倒计时完成，返回成功结果
        goal_handle.succeed()
        result.success = True
        result.message = f'倒计时完成！从 {target_number} 倒数到 0'

        self.get_logger().info(result.message)

        return result

def main(args=None):
    """主函数"""
    # 初始化ROS2 Python接口
    rclpy.init(args=args)

    # 创建动作服务端节点
    action_server = CountdownActionServer()

    try:
        # 循环等待节点退出
        rclpy.spin(action_server)
    except KeyboardInterrupt:
        pass
    finally:
        # 销毁节点
        action_server.destroy_node()
        # 关闭ROS2 Python接口
        rclpy.shutdown()

if __name__ == '__main__':
    main()

（七）编写 Action Client 节点

在 learn_action 功能包下创建 action_client.py 文件：

python

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

import rclpy                                       # ROS2 Python接口库
from rclpy.node import Node                        # ROS2 节点类
from rclpy.action import ActionClient              # ROS2 动作客户端类

# 导入自定义动作接口
from action_interfaces.action import Countdown

class CountdownActionClient(Node):
    """
    倒计时动作客户端节点
    发送倒计时目标请求，接收反馈和结果
    """

    def __init__(self, node_name='countdown_action_client'):
        super().__init__(node_name)

        # 创建动作客户端
        # 参数：节点对象、动作类型、动作名称
        self._action_client = ActionClient(
            self,
            Countdown,      # 动作接口类型
            'countdown'     # 动作名称（与服务端一致）
        )

        self.get_logger().info('倒计时动作客户端已启动！')

    def send_goal(self, target_number):
        """
        发送目标请求

        参数:
            target_number: 目标数字，从这里开始倒计时
        """
        self.get_logger().info(f'准备发送目标: 从 {target_number} 开始倒计时')

        # 等待动作服务端启动
        # timeout_sec: 等待超时时间（秒）
        if not self._action_client.wait_for_server(timeout_sec=5.0):
            self.get_logger().error('动作服务端未启动！')
            return

        self.get_logger().info('动作服务端已连接')

        # 创建目标请求对象
        goal_msg = Countdown.Goal()
        goal_msg.target_number = target_number

        self.get_logger().info(f'发送目标请求...')

        # 异步发送目标请求
        # 返回一个 Future 对象，用于获取结果
        send_goal_future = self._action_client.send_goal_async(
            goal_msg,
            feedback_callback=self.feedback_callback  # 设置反馈回调函数
        )

        # 注册结果回调（当目标被接受或拒绝时触发）
        send_goal_future.add_done_callback(self.goal_response_callback)

    def goal_response_callback(self, future):
        """
        目标响应回调函数
        当服务端接受或拒绝目标时被调用

        参数:
            future: 包含目标句柄的Future对象
        """
        goal_handle = future.result()

        # 检查目标是否被接受
        if not goal_handle.accepted:
            self.get_logger().error('目标被拒绝')
            return

        self.get_logger().info('目标已被接受')

        # 异步获取结果
        get_result_future = goal_handle.get_result_async()

        # 注册结果回调
        get_result_future.add_done_callback(self.get_result_callback)

    def feedback_callback(self, feedback_msg):
        """
        反馈回调函数
        当收到服务端的反馈时被调用

        参数:
            feedback_msg: 反馈消息对象
        """
        # 获取反馈数据
        feedback = feedback_msg.feedback
        self.get_logger().info(f'收到反馈: 当前数字 {feedback.current_number}')

    def get_result_callback(self, future):
        """
        结果回调函数
        当任务完成并返回结果时被调用

        参数:
            future: 包含结果的Future对象
        """
        # 获取结果数据
        result = future.result().result

        self.get_logger().info(f'任务完成！')
        self.get_logger().info(f'成功: {result.success}')
        self.get_logger().info(f'消息: {result.message}')

        # 关闭节点
        rclpy.shutdown()

def main(args=None):
    """主函数"""
    # 初始化ROS2 Python接口
    rclpy.init(args=args)

    # 创建动作客户端节点
    action_client = CountdownActionClient()

    try:
        # 发送目标请求：从数字5开始倒计时
        action_client.send_goal(target_number=5)

        # 循环等待节点退出
        rclpy.spin(action_client)
    except KeyboardInterrupt:
        pass
    except rclpy.exceptions.RCLpyError:
        # 正常退出时会有异常，忽略即可
        pass
    finally:
        # 销毁节点
        action_client.destroy_node()

if __name__ == '__main__':
    main()

（八）修改 setup.py

在 setup.py 中添加入口点配置：

python

from setuptools import setup

package_name = 'learn_action'

setup(
    name=package_name,
    version='0.0.0',
    packages=[package_name],
    data_files=[
        ('share/ament_index/resource_index_packages',
            ['resource/' + package_name]),
        ('share/' + package_name, ['package.xml']),
    ],
    install_requires=['setuptools'],
    zip_safe=True,
    maintainer='user',
    maintainer_email='user@todo.todo',
    description='ROS2 action learning package',
    license='TODO: License declaration',
    tests_require=['pytest'],
    entry_points={
        'console_scripts': [
            'countdown_server = learn_action.action_server:main',
            'countdown_client = learn_action.action_client:main',
        ],
    },
)

（九）编译运行与验证

1. 编译功能包

返回工作空间根目录进行编译：

bash

cd ~/ros2_ws
colcon build
source install/setup.bash

2. 运行动作服务端

打开第一个终端：

bash

cd ~/ros2_ws
source install/setup.bash
ros2 run learn_action countdown_server

3. 运行动作客户端

打开第二个终端：

bash

cd ~/ros2_ws
source install/setup.bash
ros2 run learn_action countdown_client

客户端发送目标请求后，服务端开始执行倒计时任务，期间持续发送反馈，最终返回结果。

（十）使用命令行工具验证 Action

1. 查看动作列表

bash

ros2 action list

2. 查看动作信息

bash

ros2 action info /countdown

3. 通过命令行发送动作目标

bash

# 格式: ros2 action send_goal 动作名 动作类型 "{目标数据}"
ros2 action send_goal /countdown action_interfaces/action/Countdown "{target_number: 3}"

添加 --feedback 参数可以查看反馈信息：

bash

ros2 action send_goal /countdown action_interfaces/action/Countdown "{target_number: 3}" --feedback

动作通信总结

动作通信的关键要点

说明核心概念
动作适合长时间运行的任务，提供目标-反馈-结果机制
使用 .action 文件定义动作接口
服务端通过 ActionServer 创建，处理目标和发送反馈
客户端通过 ActionClient 创建，发送目标和接收结果
使用 ReentrantCallbackGroup 防止回调阻塞

动作与服务的选择

场景	推荐方式	原因
快速查询/计算（毫秒级）	服务	一次请求-响应即可完成
长时间任务（秒级以上）	动作	需要进度反馈和取消能力
需要实时监控进度	动作	支持持续反馈机制
需要中断执行	动作	支持目标取消功能

ROS2 动作通信 ​

动作概念 ​

1. 什么是动作（Action） ​

2. Action 与 Topic、Service 的区别 ​

话题（Topic） ​

服务（Service） ​

动作（Action） ​

3. Action 的客户端与服务端 ​

4. .action 文件结构 ​

学习内容：动作实践 ​

（一）创建 Action 接口包 ​

（二）创建 action 目录 ​

（三）编写 Countdown.action 自定义动作接口 ​

（四）修改 CMakeLists.txt 和 package.xml ​

1. 修改 CMakeLists.txt ​

2. 修改 package.xml ​

（五）创建使用动作接口的 Python 功能包 ​

（六）编写 Action Server 节点 ​

（七）编写 Action Client 节点 ​

（八）修改 setup.py ​

（九）编译运行与验证 ​

1. 编译功能包 ​

2. 运行动作服务端 ​

3. 运行动作客户端 ​

（十）使用命令行工具验证 Action ​

1. 查看动作列表 ​

2. 查看动作信息 ​

3. 通过命令行发送动作目标 ​

动作通信总结 ​

动作通信的关键要点 ​

动作与服务的选择 ​

ROS2 动作通信

动作概念

1. 什么是动作（Action）

2. Action 与 Topic、Service 的区别

话题（Topic）

服务（Service）

动作（Action）

3. Action 的客户端与服务端

4. .action 文件结构

学习内容：动作实践

（一）创建 Action 接口包

（二）创建 action 目录

（三）编写 Countdown.action 自定义动作接口

（四）修改 CMakeLists.txt 和 package.xml

1. 修改 CMakeLists.txt

2. 修改 package.xml

（五）创建使用动作接口的 Python 功能包

（六）编写 Action Server 节点

（七）编写 Action Client 节点

（八）修改 setup.py

（九）编译运行与验证

1. 编译功能包

2. 运行动作服务端

3. 运行动作客户端

（十）使用命令行工具验证 Action

1. 查看动作列表

2. 查看动作信息

3. 通过命令行发送动作目标

动作通信总结

动作通信的关键要点

动作与服务的选择